/**
 * Copyright (C) 2009+ Spark Zheng
 *
 * @file        src/filt_common.c 
 * @date        2010-05-07
 * @brief	general_datastream
 * @see         dserver
 * @history
 */

#include <dlib/list.h>
#include <dlib/str.h>

#include "dserver.h"
#include "reactor.h"
#include "dserver_log.h"
#include "dserver_errno.h"
#include "filter.h"
#include "pipeline.h"
#include "filt_types.h"

/** global dserver context struct defined in dserver.c[line 38] */
extern dserver_t dc;

int filter_get_ids_by_name(const char *name, uint16_t *fid, uint16_t *pid)
{
	filter_node_t *tpos = NULL;
	list_node_t *pos = NULL;
	dserver_filter_t *filters = dc.filters;

	LIST_FOREACH_ENTRY(tpos, pos, &filters->f_head, f_node) {
		if (str_compare(name, tpos->f_name) == 0) {
			*fid = tpos->f_settings.f_filter_id; 
			*pid = tpos->f_settings.f_pipeline_id;
			return RET_SUCCESS;
		}
	}
	return -E_ITEM_NOTEXIST;
}


char *filter_get_name_by_fid(uint16_t fid)
{
	filter_node_t *tpos = NULL;
	list_node_t *pos = NULL;
	dserver_filter_t *filters = dc.filters;

	LIST_FOREACH_ENTRY(tpos, pos, &filters->f_head, f_node) {
		if (tpos->f_settings.f_filter_id == fid) {
			return tpos->f_name;
		}
	}
	return NULL;
}

filter_node_t *filter_get_by_fid(uint16_t fid)
{
	filter_node_t *tpos = NULL;
	list_node_t *pos = NULL;
	dserver_filter_t *filters = dc.filters;

	LIST_FOREACH_ENTRY(tpos, pos, &filters->f_head, f_node) {
		if (tpos->f_settings.f_filter_id == fid) {
			return tpos;
		}
	}
	return NULL;
}

pipeline_node_t *pipeline_get_by_fid(uint16_t fid)
{
	uint16_t pid = 0;
	filter_node_t *tpos = NULL;
	list_node_t *pos = NULL;
	dserver_filter_t *filters = dc.filters;
	dserver_pipeline_t *pipelines = dc.pipelines;

	LIST_FOREACH_ENTRY(tpos, pos, &filters->f_head, f_node) {
		if (tpos->f_settings.f_filter_id == fid) {
			pid = tpos->f_settings.f_pipeline_id;
			if (pid < pipelines->p_max_id)
				return pipelines->p_nodes + pid;
			else
				return NULL;
		}
	}
	return NULL;
}

filter_node_t *filter_get_by_name(const char *name)
{
	filter_node_t *tpos = NULL;
	list_node_t *pos = NULL;
	dserver_filter_t *filters = dc.filters;

	LIST_FOREACH_ENTRY(tpos, pos, &filters->f_head, f_node) {
		if (str_compare(name, tpos->f_name) == 0) {
			return tpos;
		}
	}
	return NULL;
}

pipeline_node_t *pipeline_get_by_name(const char *name)
{
	uint16_t pid = 0;
	filter_node_t *tpos = NULL;
	list_node_t *pos = NULL;
	dserver_filter_t *filters = dc.filters;
	dserver_pipeline_t *pipelines = dc.pipelines;

	LIST_FOREACH_ENTRY(tpos, pos, &filters->f_head, f_node) {
		if (str_compare(name, tpos->f_name) == 0) {
			pid = tpos->f_settings.f_pipeline_id;
			if (pid < pipelines->p_max_id)
				return pipelines->p_nodes + pid;
			else
				return NULL;
		}
	}
	return NULL;
}


/** pipeline core APIs */
int pipeline_dispatch(pipeline_node_t *pnode, filter_node_t *fnode, pipeline_data_t *pdata)
{
	int ret;

	if (fnode == NULL || pdata == NULL) {
		return -E_ARGS_ILLEGAL;
	}

	dlib_tpool_t *tpool = fnode->f_tpool;

	/* force to set pipeline flag to BEGIN avoid protocol module abuse IGNORE */
	pdata->p_prot_flag = PROT_PIPELINE_BEGIN;

	if (pnode == NULL) {
		pnode = dc.pipelines->p_nodes + fnode->f_settings.f_pipeline_id;
	}

	pthread_mutex_lock(&pdata->p_mutex);

#ifdef DSERVER_DATASTREAM_NEED_DLOG
	DLOG_DEBUG(LOG_MAIN, "pipeline_dispatch to filter[%s:%d:%d].", fnode->f_name,
			fnode->f_settings.f_pipeline_id, fnode->f_settings.f_filter_id);
#endif

	/* deliver data to pipeline */
	ret = dlib_tpool_execute(tpool, fnode->f_settings.f_priority, NULL, (void *)pdata);
	if (ret >= 0) {
		/* wait for result */
		while (pdata->p_prot_flag != PROT_PIPELINE_END) {
			/* TODO: need pthread_cond_timedwait */
			pthread_cond_wait(&pdata->p_cond, &pdata->p_mutex);
		}
#ifdef DSERVER_DATASTREAM_NEED_DLOG
		DLOG_DEBUG(LOG_MAIN, "pipeline_dispatch from filter[%s:%d:%d].",
				pdata->p_submit_fnode->f_name,
				pdata->p_submit_fnode->f_settings.f_pipeline_id,
				pdata->p_submit_fnode->f_settings.f_filter_id);
#endif
	} else {
#ifdef DSERVER_DATASTREAM_NEED_DLOG
		DLOG_ERR(LOG_MAIN, "pipeline_dispatch error.");
#endif
	}
	pthread_mutex_unlock(&pdata->p_mutex);

	return ret;
}

int pipeline_request(pipeline_node_t *pnode, filter_node_t *fnode, pipeline_data_t *pdata)
{
	if (fnode == NULL || pdata == NULL) {
		return -E_ARGS_ILLEGAL;
	}

	dlib_tpool_t *tpool = fnode->f_tpool;
	if (pnode == NULL) {
		pnode = dc.pipelines->p_nodes + fnode->f_settings.f_pipeline_id;
	}

#ifdef DSERVER_DATASTREAM_NEED_DLOG
	DLOG_DEBUG(LOG_MAIN, "pipeline_request to filter[%s:%d:%d].", fnode->f_name,
			fnode->f_settings.f_pipeline_id, fnode->f_settings.f_filter_id);
#endif

	return dlib_tpool_execute(tpool, fnode->f_settings.f_priority, NULL, (void *)pdata);
}

int pipeline_deliver_next(filter_node_t *fnode, pipeline_data_t *pdata)
{
	if (fnode == NULL || pdata == NULL) {
		return -E_ARGS_ILLEGAL;
	}
	pipeline_node_t *pnode = dc.pipelines->p_nodes + fnode->f_settings.f_pipeline_id;
	filter_node_t *fnext = NULL;

	fnext = LIST_NEXT_ENTRY(&pnode->p_filter_head, &fnode->p_node, filter_node_t, p_node);
	if (fnext == NULL) { /* the last entry, need to submit */
		return pipeline_submit(fnode, pdata);
	} 

	/* pass data to next filter */
	dlib_tpool_t *tpool = fnext->f_tpool;

#ifdef DSERVER_DATASTREAM_NEED_DLOG
	DLOG_DEBUG(LOG_MAIN, "pipeline_deliver_next from filter[%s:%d:%d] to filter[%s:%d:%d].",
			fnode->f_name, fnode->f_settings.f_pipeline_id,
			fnode->f_settings.f_filter_id, fnext->f_name,
			fnext->f_settings.f_pipeline_id, fnext->f_settings.f_filter_id);
#endif

	return dlib_tpool_execute(tpool, fnext->f_settings.f_priority, NULL, (void *)pdata);
}

int pipeline_deliver(filter_node_t *fnode, pipeline_data_t *pdata)
{
	if (fnode == NULL || pdata == NULL) {
		return -E_ARGS_ILLEGAL;
	}
	dlib_tpool_t *tpool = fnode->f_tpool;

#ifdef DSERVER_DATASTREAM_NEED_DLOG
	DLOG_DEBUG(LOG_MAIN, "pipeline_deliver to filter[%s:%d:%d].", fnode->f_name,
			fnode->f_settings.f_pipeline_id, fnode->f_settings.f_filter_id);
#endif

	return dlib_tpool_execute(tpool, fnode->f_settings.f_priority, NULL, (void *)pdata);
}

int pipeline_submit(filter_node_t *fnode, pipeline_data_t *pdata)
{
	if (fnode == NULL || pdata == NULL) {
		return -E_ARGS_ILLEGAL;
	}

	pipeline_node_t *pnode = dc.pipelines->p_nodes + fnode->f_settings.f_pipeline_id;
	pdata->p_submit_fnode = fnode;
	pdata->p_submit_pnode = pnode;

	if (pdata->p_prot_flag == PROT_PIPELINE_IGNORE) {
		return -E_PIPELINE_ASYNC;
	}

#ifdef DSERVER_DATASTREAM_NEED_DLOG
	DLOG_DEBUG(LOG_MAIN, "pipeline_submit from filter[%s:%d:%d].", fnode->f_name,
			fnode->f_settings.f_pipeline_id, fnode->f_settings.f_filter_id);
#endif

	/* signal to protocol-worker */
	pthread_mutex_lock(&pdata->p_mutex);
	pdata->p_prot_flag = PROT_PIPELINE_END;
	pthread_cond_signal(&pdata->p_cond);
	pthread_mutex_unlock(&pdata->p_mutex);

	return RET_SUCCESS;
}

int pipeline_submit_async(filter_node_t *fnode, pipeline_data_t *pdata)
{
	if (fnode == NULL || pdata == NULL) {
		return -E_ARGS_ILLEGAL;
	}

	int pipe_size = 0;
	char pipe_buf[REACTOR_PIPE_BUF_SIZE] = {0};

	pipe_size = snprintf(pipe_buf, REACTOR_PIPE_BUF_SIZE, "%ld", (long)(pdata->p_data));
	if (write(dc.reactor->r_done_pipe[1], pipe_buf, pipe_size) < 0) { 
#ifdef DSERVER_DATASTREAM_NEED_DLOG
		DLOG_ERR(LOG_MAIN, "pipeline_submit_async error, lost something[%s:%s]",
				fnode->f_name, strerror(errno));
#endif
		return -E_PIPELINE_ASYNC;
	}

#ifdef DSERVER_DATASTREAM_NEED_DLOG
	DLOG_DEBUG(LOG_MAIN, "pipeline_submit_async from filter[%s:%d:%d].", fnode->f_name,
			fnode->f_settings.f_pipeline_id, fnode->f_settings.f_filter_id);
#endif

	return RET_SUCCESS;
}

int pipeline_result(pipeline_data_t *pdata)
{
	if (pdata == NULL) {
		return -E_ARGS_ILLEGAL;
	}

	/* wait for result */
	pthread_mutex_lock(&pdata->p_mutex);
	while (pdata->p_prot_flag != PROT_PIPELINE_END) {
		/* TODO: need pthread_cond_timedwait */
		pthread_cond_wait(&pdata->p_cond, &pdata->p_mutex);
	}

#ifdef DSERVER_DATASTREAM_NEED_DLOG
	DLOG_DEBUG(LOG_MAIN, "pipeline_result from filter[%s:%d:%d].",
			pdata->p_submit_fnode->f_name,
			pdata->p_submit_fnode->f_settings.f_pipeline_id,
			pdata->p_submit_fnode->f_settings.f_filter_id);
#endif
	pthread_mutex_unlock(&pdata->p_mutex);

	return RET_SUCCESS;
}
